Python middleware for generate() — use=[...]

Adds a middleware system for Python that lets you intercept and wrap generate() calls at three granularities: the full generate iteration, each model API call, and each tool execution. Replaces the old ModelMiddleware abstraction.

What it does

Middleware is applied per-generate() call via a use=[...] parameter. Each entry in the list wraps the call in a chain — first entry is outermost (runs first/last). Three hooks are available:

• wrap_generate — wraps each iteration of the tool loop (model call + tool resolution). Runs once per agentic turn.
• wrap_model — wraps each raw model API call. Use for retry, fallback, logging latency.
• wrap_tool — wraps each individual tool execution. Use for approval gates, sandboxing, error enrichment.
• tools() — contribute extra tools dynamically per generate() call (e.g. skills libraries, sandboxed filesystem ops). Tools are scoped to the call and don't pollute the root registry.

Defining and registering middleware inline (app developers)

Subclass BaseMiddleware (a Pydantic model) and decorate it using the @ai.middleware decorator on your Genkit instance. This automatically registers the middleware with the registry so that it is discoverable by the Dev UI and usable across your app:

import time

from genkit import Genkit
from genkit.middleware import BaseMiddleware, ModelHookParams
from genkit.plugins.middleware import Retry

ai = Genkit()

@ai.middleware(name='latency_logger', description='Logs model latency')
class LatencyLogger(BaseMiddleware):
    prefix: str = '[trace]'

    async def wrap_model(self, params: ModelHookParams, next_fn):
        t = time.monotonic()
        resp = await next_fn(params)
        print(f'{self.prefix} model call took {time.monotonic() - t:.3f}s')
        return resp

response = await ai.generate(
    model='googleai/gemini-flash-latest',
    prompt='Hello',
    use=[
        Retry(max_retries=5),
        LatencyLogger(prefix='[myapp]'),
    ],
)

Dev UI integration

Any middleware registered via plugins or decorated with @ai.middleware is automatically available in the Dev UI — no extra registration steps needed.

Once registered, the middleware shows up on the Model Runner page in the Dev UI, where you can mix-and-match middleware and set config values interactively. When you run a generate call from there, the Dev UI passes a MiddlewareRef — a name plus a config dict — into generate_action. The framework resolves that ref against the registry, instantiates the middleware class with the provided config (cls(**config)), and runs the chain exactly as it would inline.

Pre-packaging middleware through a plugin (plugin authors)

Use new_middleware to build a MiddlewareDesc from a BaseMiddleware subclass, then wrap them with middleware_plugin to produce a standard Plugin for plugins=[...]. Note that plugin middleware classes do not require decorators:

mylib/middleware.py (plugin author):

from genkit.middleware import BaseMiddleware, ModelHookParams
from genkit.plugin_api import new_middleware, middleware_plugin

class Retry(BaseMiddleware):
    max_retries: int = 3

    async def wrap_model(self, params: ModelHookParams, next_fn):
        for attempt in range(self.max_retries + 1):
            try:
                return await next_fn(params)
            except Exception:
                if attempt == self.max_retries:
                    raise

class Fallback(BaseMiddleware):
    models: list[str] = []

    async def wrap_model(self, params: ModelHookParams, next_fn):
        ...  # try primary, then each fallback model

def my_middleware_plugin():
    return middleware_plugin([
        new_middleware(Retry, name='retry', description='Retries model calls on transient failures'),
        new_middleware(Fallback, name='fallback'),
    ])